package loadData

import (
	"fmt"
	"github.com/google/uuid"
	cp "db2s/componentInit"
	"db2s/global"
	"db2s/limit"
	mq "db2s/topic-mq"
	"reflect"
	"runtime"
	"strings"
	"sync/atomic"
)

func loadDataMQListenInitFunc(t, p int) map[cp.CheckMode][]*cp.MQTopicListeners {
	MQListenersTable := map[Topic]*cp.MQTopicListeners{
		loadOutputIsSegmentQuery:       cp.NewMQTopicListeners(cp.Topic(loadOutputIsSegmentQuery), t, 1, loadOutputIsSegmentQueryListeners, 5, 10),
		loadOutputMissSegmentQuery:     cp.NewMQTopicListeners(cp.Topic(loadOutputMissSegmentQuery), t, 1, loadOutputMissSegmentQueryListeners, 5, 10),
		loadOutputIndexConcurrentQuery: cp.NewMQTopicListeners(cp.Topic(loadOutputIndexConcurrentQuery), t*p*2, 1000, loadOutputIndexConcurrentQueryListeners, 5, 20),
		loadOutputTableSql:             cp.NewMQTopicListeners(cp.Topic(loadOutputTableSql), t*p*2, 1000, loadOutputTableSqlListeners, 1, 30),
		loadOutputQueryTableData:       cp.NewMQTopicListeners(cp.Topic(loadOutputQueryTableData), t*p*2, 1000, loadOutputQueryTableDataListeners, 5, 41),
		loadOutputWriteData:            cp.NewMQTopicListeners(cp.Topic(loadOutputWriteData), t*p*2, 1000, loadOutputWriteDataListeners, 5, 42),
		loadInputFileData:              cp.NewMQTopicListeners(cp.Topic(loadInputFileData), t*p*2, 1000, loadInputFileDataListeners, 5, 43),
		loadInputGenerlInsert:          cp.NewMQTopicListeners(cp.Topic(loadInputGenerlInsert), t*p*2, 1000, loadInputGenerlInsertListeners, 5, 42),
		loadInputWriteInsert:           cp.NewMQTopicListeners(cp.Topic(loadInputWriteInsert), t*p*2, 1000, loadInputWriteInsertListeners, 5, 43),
	}
	return map[cp.CheckMode][]*cp.MQTopicListeners{
		"load": {
			MQListenersTable[loadOutputIsSegmentQuery],
			MQListenersTable[loadOutputMissSegmentQuery],
			MQListenersTable[loadOutputIndexConcurrentQuery],
			MQListenersTable[loadOutputTableSql],
			MQListenersTable[loadOutputQueryTableData],
			MQListenersTable[loadOutputWriteData],
			MQListenersTable[loadInputFileData],
			MQListenersTable[loadInputGenerlInsert],
			MQListenersTable[loadInputWriteInsert],
		},
	}
}
func topicCreateInit(checkmod string, t, p int) (m map[cp.CheckMode][]*cp.MQTopicListeners) {
	m = make(map[cp.CheckMode][]*cp.MQTopicListeners)
	if funcName, ok := MqListenerSInitFunc[checkmod]; ok {
		if funcName.Kind() == reflect.Func {
			m1 := funcName.Call([]reflect.Value{
				reflect.ValueOf(t),
				reflect.ValueOf(p),
			})
			for i := 0; i < len(m1); i++ {
				m = m1[i].Interface().(map[cp.CheckMode][]*cp.MQTopicListeners)
			}
		}
	}
	return m
}
func GetCheckToTopicMQListeners() map[cp.CheckMode][]*cp.MQTopicListeners {
	var (
		t = GlobalPConfigs.rules.Mtc
		r = GlobalPConfigs.rules.Mcc
	)
	return topicCreateInit(GlobalPConfigs.rules.CheckMode, t, r)
}

func (context *SchedulePlanContext) topicEndAdd(index CheckMode, topic Topic) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndAdd(cp.Topic(topic))
}
func (context *SchedulePlanContext) topicEndMonit(index CheckMode, topic Topic) int64 {
	return context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndMonit(cp.Topic(topic))
}
func (context *SchedulePlanContext) topicEndDelete(index CheckMode, topic Topic) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndDelete(cp.Topic(topic))
}

func (context *SchedulePlanContext) topicEndDeleteNumber(index CheckMode, topic Topic, n int64) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndDeleteNumber(cp.Topic(topic), n)
}

func sendMsg(check CheckMode, topic Topic, tableName string, msg map[string]any) bool {
	var (
		outFay string
		event  = "[sendMsg]"
		logSeq int64
		vlog   string
	)
	pc, _, _, _ := runtime.Caller(1)
	callingFunc := runtime.FuncForPC(pc)
	callFile, callLine := callingFunc.FileLine(pc)
	callFunc := callingFunc.Name()
	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
	vlog = fmt.Sprintf("(%d) %v %v Start sending topic messages, topic is {%v].", logSeq, callFuncInfo, event, topic)
	WLog.Debug(vlog)
	PlanContext.topicEndAdd(check, topic)
	if v, ok := msg["logSeq"]; !ok {
		vlog = fmt.Sprintf("(%d) %v %v The logseq value was not obtained and the sending failed.", logSeq, callFuncInfo, event)
		WLog.Error(vlog)
		return false
	} else {
		logSeq = v.(int64)
	}
	if err := producer.Send(mq.Message{
		Topic:      string(topic),
		Body:       []byte(tableName),
		Properties: msg,
	}); err != nil {
		WLog.Error(fmt.Sprintf("(%d) %v %s Sending data check topic %v message failed", logSeq, callFuncInfo, event, topic))
		return false
	}
	if v, ok := msg["outFay"]; ok {
		outFay = v.(string)
	}
	if strings.EqualFold(outFay, "bar") {
		PlanContext.taskBarEndTopicAdd(msg)
	}
	vlog = fmt.Sprintf(fmt.Sprintf("(%d) %v %s Sending data check topic %v message successfully", logSeq, callFuncInfo, event, topic))
	WLog.Debug(vlog)
	return true
}

type FatherSon struct {
	FatherTags    map[string]any
	FatherSonTags map[string][]string
}
type SendMsgInfo struct {
	CheckMod  CheckMode
	Topic     Topic
	TableName string
}
type chanSendMsg struct {
	Object           string
	SendData         chan any
	SendMsgInfo      SendMsgInfo
	ExecFunc         func()
	SpeedLimitSwitch bool
	SpeedLimit       *limit.RateLimiter
	Status           *int64
	SendMsg          sendTopicF
	FatherSon        FatherSon
	SendMsgEnd       sendEndDispos
}

// 循环读取管道并发送消息
func chanMonitorSendMsg(m *chanSendMsg) bool {
	var (
		uid    uuid.UUID
		vlog   string
		event  = "[chanMonitorSendMsg]"
		logSeq int64
	)
	pc, _, _, _ := runtime.Caller(1)
	callingFunc := runtime.FuncForPC(pc)
	callFile, callLine := callingFunc.FileLine(pc)
	callFunc := callingFunc.Name()
	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
	vlog = fmt.Sprintf("(%d) %v %v Start reading chan and send topic messages.", logSeq, callFuncInfo, event)
	WLog.Debug(vlog)
	if v, ok := m.FatherSon.FatherTags["logSeq"]; !ok {
		vlog = fmt.Sprintf("(%d) %v %v The logseq value was not obtained and the sending failed.", logSeq, callFuncInfo, event)
		WLog.Error(vlog)
		return false
	} else {
		logSeq = v.(int64)
	}
	var emptyS = global.IsEmptyStruct(m.SpeedLimit)
	for {
		if m.SpeedLimitSwitch {
			if atomic.LoadInt64(m.Status) == 1 {
				if !emptyS && !m.SpeedLimit.TryAcquire() {
					continue
				}
			}
		}
		select {
		case c, ok := <-m.SendData:
			if !ok {
				m.SendMsgEnd()
				return true
			} else {
				if m.ExecFunc != nil {
					m.ExecFunc()
				}
				var ms = make(map[string]any)
				uid = uuid.New()
				for k, v := range m.FatherSon.FatherTags {
					if strings.EqualFold(k, "sendMsgKey") {
						ms[v.(string)] = c
					} else {
						ms[k] = v
					}
				}
				//标签生成：
				if v1, ok1 := ms["label"]; ok1 { //标签标识
					for k2 := range v1.(map[string]any) { //循环集合
						if strings.EqualFold(k2, "no") { //first label
							ms["label"] = map[string]any{uid.String(): []string{}}
							m.FatherSon.FatherSonTags[uid.String()] = []string{}
						} else if len(k2) > 0 { //子级标签
							if v4, ok4 := m.FatherSon.FatherSonTags[k2]; ok4 {
								v4 = append(v4, uid.String())
								m.FatherSon.FatherSonTags[k2] = v4
							} else {
								m.FatherSon.FatherSonTags[k2] = []string{uid.String()}
							}
							ms["label"] = map[string]any{k2: uid.String()}
						}
					}
				}
				if !m.SendMsg(m.SendMsgInfo.CheckMod, m.SendMsgInfo.Topic, m.SendMsgInfo.TableName, ms) {
					WLog.Error(fmt.Sprintf("(%d) %v %s Sending data check topic %v message failed", logSeq, callFuncInfo, event, m.SendMsgInfo.Topic))
					return false
				}
			}
		}
	}
}
func deleteTopic(topicMsg map[string]any) {
	var (
		logSeq int64
		event  string
		dTopic Topic
		outFay = "bar"
		vlog   string
	)
	pc, _, _, _ := runtime.Caller(1)
	callingFunc := runtime.FuncForPC(pc)
	callFile, callLine := callingFunc.FileLine(pc)
	callFunc := callingFunc.Name()
	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
	vlog = fmt.Sprintf("(%d) %v %v Start delete chan topic messages.", logSeq, callFuncInfo, event)
	WLog.Debug(vlog)
	if v, ok := topicMsg["logSeq"]; ok {
		logSeq = v.(int64)
	}
	if v, ok := topicMsg["event"]; ok {
		event = v.(string)
	}
	if v, ok := topicMsg["dtopic"]; ok {
		dTopic = v.(Topic)
	}
	if v, ok := topicMsg["outFay"]; ok {
		outFay = v.(string)
	}
	PlanContext.topicEndDelete(getCheckMod(), dTopic)
	if strings.EqualFold(outFay, "bar") {
		PlanContext.TaskBarEndTopicDel(topicMsg)
	}
	WLog.Debug(fmt.Sprintf("(%d) %v %s Delete data check topic %v successfully.", logSeq, callFuncInfo, event, dTopic))
}
